package com.jnpc.spark.project.dao
import com.jnpc.spark.project.domain.{FamenGuzhangCount, FamenGuzhangItems}
import com.jnpc.spark.project.utils.HBaseUtils
import org.apache.hadoop.hbase.client.{Get, Put}
import org.apache.hadoop.hbase.util.Bytes

import scala.collection.mutable.ListBuffer

/**
  * 数据访问层
  */
object FaMenGuzhangCountDAO {

  val tableName = "jnpc_crs_famenguzhangcount"
  val cf = "info"
  val qualifer = "guzhang_count"
  val qualifer2 = "guzhang_items"

  /**
    * 保存数据到HBase
    * @param list  FamenGuzhangCount
    */
  def save(list: ListBuffer[FamenGuzhangCount]): Unit = {

    val table = HBaseUtils.getInstance().getTable(tableName)

    for(ele <- list) {
      table.incrementColumnValue(Bytes.toBytes(ele.day_gztype),
        Bytes.toBytes(cf),
        Bytes.toBytes(qualifer),
        ele.count)

    }

  }
  /**
    * 保存数据到HBase
    * @param list  FamenGuzhangItems
    */
  def saveItems(list: ListBuffer[FamenGuzhangItems]): Unit = {

    val table = HBaseUtils.getInstance().getTable(tableName)

    for(ele <- list) {
      val p = new Put(Bytes.toBytes(ele.day_gztype))
       p.add(Bytes.toBytes(cf), Bytes.toBytes(qualifer2),Bytes.toBytes(ele.items))
       table.put(p)
    }

  }

  def saveItems2(list: ListBuffer[FamenGuzhangItems]): Unit = {

    val table = HBaseUtils.getInstance().getTable(tableName)

    for(ele <- list) {
      val get = new Get(Bytes.toBytes(ele.day_gztype))
      val value = table.get(get).getValue(cf.getBytes, qualifer2.getBytes)
      if(value == null) {
        val p = new Put(Bytes.toBytes(ele.day_gztype))
        p.add(Bytes.toBytes(cf), Bytes.toBytes(qualifer2),Bytes.toBytes(ele.items))
        table.put(p)
      }else{
        val p = new Put(Bytes.toBytes(ele.day_gztype))
        p.add(Bytes.toBytes(cf), Bytes.toBytes(qualifer2),Bytes.toBytes(value+","+ele.items))
        table.put(p)
      }

    }

  }


  /**
    * 根据rowkey查询值
    */
  def count(day_gztype: String):Long = {
    val table = HBaseUtils.getInstance().getTable(tableName)

    val get = new Get(Bytes.toBytes(day_gztype))
    val value = table.get(get).getValue(cf.getBytes, qualifer.getBytes)

    if(value == null) {
      0L
    }else{
      Bytes.toLong(value)
    }
  }

  def main(args: Array[String]): Unit = {


    val list = new ListBuffer[FamenGuzhangCount]
    list.append(FamenGuzhangCount("20171111_8",8))
    list.append(FamenGuzhangCount("20171111_9",9))
    list.append(FamenGuzhangCount("20171111_1",100))

    save(list)

    println(count("20171111_8") + " : " + count("20171111_9")+ " : " + count("20171111_1"))
  }

}
